package com.shujia.opt

import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object Code02SparkFilterData {
  def main(args: Array[String]): Unit = {
    /**
     * Spark数据倾斜：
     * 表现：在Job任务执行时，部分的Task一直处于运行状态，而其他的Task已经处理完成
     * 当通过执行过程查看部分Task出现数据倾斜，怎么判断是有哪些Key导致的？
     * 对数据进行抽样 -> 之后再去对Key进行分组统计，查看哪些key占比较多 并进行分析
     *
     * 方式1：对数据进行过滤操作
     */


    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("spark")
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext
    val data: RDD[String] = sc.textFile("spark_code/data/word/*")

    //    data
    //      .flatMap(_.split(","))
    //      .map((_, 1))
    //      .groupBy(_._1)
    //      .mapValues(_.size)
    //      .foreach(println)
    val mapRDD: RDD[(String, Int)] = data
      .flatMap(_.split(","))
      .map((_, 1))

    mapRDD.cache()

    val topNMap: Map[String, Int] = mapRDD
      .sample(false, 0.1)
      .groupBy(_._1)
      .mapValues(_.size)
      .map(x => (x._2, x._1))
      .top(3)
      .map {
        case (value, key) => (key, value)
      }
      .toMap // 对数据进行采样，再统计个数，之后再转换KeyValue位置，取前3

    val broadcastMap: Broadcast[Map[String, Int]] = sc.broadcast(topNMap)


    val groupByRDD: RDD[(String, Iterable[(String, Int)])] = mapRDD
      .filter {
        case (key, value) => !broadcastMap.value.contains(key)
      }
      .groupBy(_._1)

    mapRDD.unpersist() // 清空缓存

    groupByRDD
      .mapValues(_.size)
      .foreach(println)


    // (NULL,23680)
    while (true) {}
  }
}
